﻿using System;
using System.Collections.Generic;
using System.Diagnostics;
using System.Linq;
using System.Net;
using System.Threading;
using Microsoft.WindowsAzure;
using Microsoft.WindowsAzure.Diagnostics;
using Microsoft.WindowsAzure.ServiceRuntime;
using Microsoft.WindowsAzure.StorageClient;
using System.IO;
using Microsoft.WindowsAzure.Search;

namespace Parser
{
    public class WorkerRole : RoleEntryPoint
    {
        // queue
        private CloudQueue parserQueue;
        private CloudQueue inverterQueue;

        // client
        private CloudBlobClient dataClient;
        private CloudBlobClient indexBlobClient;
        private CloudQueueClient queueClient;

        // container that contains the intermediate result after parsing
        private CloudBlobContainer interContainer;
        private CloudBlobContainer logContainer;

        private int docID;
        private int start;

        // Blob Stream Writer
        private BlobStream docWriter;

        // table 
        private DocinfoContext context;

        private void Initialize()
        {           
            CloudStorageAccount dataAccount = CloudStorageAccount.FromConfigurationSetting("DataConnectionString");
            CloudStorageAccount indexAccount = CloudStorageAccount.FromConfigurationSetting("IndexConnectionString");

            dataClient = dataAccount.CreateCloudBlobClient();
            indexBlobClient = indexAccount.CreateCloudBlobClient();
            queueClient = indexAccount.CreateCloudQueueClient();

            // create parser queue if not exist
            parserQueue = queueClient.GetQueueReference("parser");
            parserQueue.CreateIfNotExist();

            // container
            interContainer = indexBlobClient.GetContainerReference("temp");
            interContainer.CreateIfNotExist();
            logContainer = indexBlobClient.GetContainerReference("log");
            logContainer.CreateIfNotExist();

            // queue
            inverterQueue = queueClient.GetQueueReference("inverter");
            inverterQueue.CreateIfNotExist();

            // table context
            CloudTableClient tableClient = indexAccount.CreateCloudTableClient();
            tableClient.CreateTableIfNotExist("docinfo");
            context = new DocinfoContext(indexAccount.TableEndpoint.ToString(), indexAccount.Credentials);
        }

        public override bool OnStart()
        {
            // Set the maximum number of concurrent connections 
            ServicePointManager.DefaultConnectionLimit = 12;

            // For information on handling configuration changes
            // see the MSDN topic at http://go.microsoft.com/fwlink/?LinkId=166357.

            #region Setup CloudStorageAccount Configuration Setting Publisher
            // This code sets up a handler to update CloudStorageAccount instances when their
            // corresponding configuration settings change in the service configuration file.
            CloudStorageAccount.SetConfigurationSettingPublisher((configName, configSetter) =>
            {
                // Provide the configSetter with the initial value
                configSetter(RoleEnvironment.GetConfigurationSettingValue(configName));
                RoleEnvironment.Changed += (s, arg) =>
                {
                    if (arg.Changes.OfType<RoleEnvironmentConfigurationSettingChange>()
                    .Any((change) => (change.ConfigurationSettingName == configName)))
                    {
                        // The corresponding configuration setting has changed,
                        // propagate the value
                        if (!configSetter(RoleEnvironment.
                        GetConfigurationSettingValue(configName)))
                        {
                            // In this case, the change to the storage account credentials in the
                            // service configuration is significant enough that the role needs
                            // to be recycled in order to use the latest settings. (for example,
                            // the endpoint has changed)
                            RoleEnvironment.RequestRecycle();
                        }
                    }
                };
            });
            #endregion

            Initialize();

            return base.OnStart();
        }

        public override void Run()
        {
            // loop to wait for messages
            while (true)
            {
                // message sent by web role to start index construction
                CloudQueueMessage m = parserQueue.GetMessage();

                if (m != null)
                {
                    Stopwatch watch = new Stopwatch();
                    watch.Start();

                    Debug.WriteLine("Start Parsing " + m.AsString);
                    parserQueue.DeleteMessage(m);

                    // message parsing
                    // <range, start number, end number>
                    String[] info = Tokenizer.CommaTokenize(m.AsString);
                    start = Convert.ToInt32(info[0]);
                    int end = Convert.ToInt32(info[1]);

                    String name = start.ToString();
                    CloudBlob blob = interContainer.GetBlobReference(name);
                    docWriter = blob.OpenWrite();

                    IEnumerable<CloudBlobContainer> containers = dataClient.ListContainers();

                    docID = 0;
                    int k = 0;
                    foreach (CloudBlobContainer container in containers)
                    {
                        IEnumerable<IListBlobItem> blobs = container.ListBlobs();
                        foreach (IListBlobItem blobItem in blobs)
                        {
                            if (start <= docID && docID <= end)
                            {
                                ParseBlob(blobItem);
                                k++;
                                if (k == 100)
                                {
                                    context.SaveChanges(System.Data.Services.Client.SaveChangesOptions.Batch);
                                    k = 0;
                                }
                            }
                            else
                            {
                                // finish
                                if (docID > end)
                                    goto Finish;
                            }
                            docID++;
                        }
                    }

                    // finish
                    Finish: Finish();
                    watch.Stop();
                    Debug.WriteLine("Parse Time: " + watch.ElapsedMilliseconds + " ms");
                    CloudBlob logBlob = logContainer.GetBlobReference("parser" + start);
                    logBlob.UploadText("Parse Time: " + watch.ElapsedMilliseconds + " ms");
                }

                Thread.Sleep(2000);
            }
        }

        private void ParseBlob(IListBlobItem blobItem)
        {
            // get the content of the blob
            String url = blobItem.Uri.AbsoluteUri;
            CloudBlob blob = dataClient.GetBlobReference(url);
            String content = blob.DownloadText();

            // compute term frquency in the blob
            Dictionary<String, int> termFreq = Analyzer.ComputeTermFreq(content);

            // compute length of the blob
            double length = 0;
            foreach (int count in termFreq.Values)
            {
                length += count * count;
            }
            length = Math.Sqrt(length);


            // write docinfo
            if (docID != start)                
                WriteData("\n", docWriter);

            // first line, write <docID>
            WriteData(docID + "\n", docWriter);

            // write docinfo to table
            context.Add(start, docID, url);


            // write term frequency
            bool firstVisit = true;
            // write term and frequency
            foreach (String term in termFreq.Keys)
            {
                double weight_tf = (double)termFreq[term] / length;
                if (firstVisit)
                    WriteData(term + "," + Math.Round(weight_tf, 5), docWriter);
                else
                    WriteData("," + term + "," + Math.Round(weight_tf, 5), docWriter);

                firstVisit = false;
            }
        }

        private void WriteData(String str, BlobStream docWriter)
        {
            byte[] buffer = System.Text.Encoding.ASCII.GetBytes(str);
            docWriter.Write(buffer, 0, buffer.Length);
        }


        // send message to inverter queue
        private void Finish()
        {
            // upload intermediate result - documents 
            docWriter.Flush();
            docWriter.Close();

            // send message to inverter
            // <start number>
            CloudQueueMessage message = new CloudQueueMessage(start.ToString());
            inverterQueue.AddMessage(message);

            // docinfo save changes
            context.SaveChanges(System.Data.Services.Client.SaveChangesOptions.Batch);

            Debug.WriteLine("************Parsing " + start +" Done************");
        }
    }
}
